package com.chencong.online.driver.dwd;

import com.chencong.online.bean.PageViewCountBean;
import com.chencong.online.bean.PageViewCountResultBean;
import com.chencong.online.bean.UserBehaviorBean;
import com.chencong.online.env.FlinkEnv;
import com.chencong.online.function.PageViewCountAggFunc;
import com.chencong.online.function.PageViewCountProcessFunc;
import com.chencong.online.function.PageViewCountWindowFunc;
import com.chencong.online.function.UserBehaviorBeanMapFunc;
import com.chencong.online.utils.TimestampTransformUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.net.URL;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Random;

/**
 * @program: user-behavior-analysis-online
 * @ClassName DwdPageView
 * @description:PV统计
 * @author: chencong
 * @create: 2021-12-28 09:59
 **/
public class DwdPageViewDriver {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = FlinkEnv.FlinkDataStreamRunEnv();
        env.setParallelism(1);
        //todo 1 source
        URL resource = DwdPageViewDriver.class.getResource("/UserBehavior.csv");
        DataStreamSource<String> inputDS = env.readTextFile(resource.getPath());
//        inputDS.print("获取原始数据");
        //todo 2 transform
        //转换层javabean并提取时间戳
        SingleOutputStreamOperator<UserBehaviorBean> userBehaviorDS = inputDS
                .map(new UserBehaviorBeanMapFunc())
                .filter(data -> "pv".equals(data.getBehavior()))
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<UserBehaviorBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<UserBehaviorBean>() {
                            @Override
                            public long extractTimestamp(UserBehaviorBean element, long recordTimestamp) {
                                return element.getTimestamp() * 1000L;
                            }
                        }));
//        userBehaviorDS.print();
        //开窗分组聚合
        SingleOutputStreamOperator<PageViewCountBean> pageViewCountAggDS = userBehaviorDS.map(new MapFunction<UserBehaviorBean, Tuple2<Integer, Long>>() {
                    @Override
                    public Tuple2<Integer, Long> map(UserBehaviorBean value) throws Exception {
                        //设计随机数key，解决数据倾斜问题（10以内的随机数），可开并行任务
                        Random random = new Random();
                        return new Tuple2(random.nextInt(10), 1L);
                    }
                })
                .keyBy(data -> data.f0)//通过生成的随机数keyBy，解决数据倾斜
                .window(TumblingEventTimeWindows.of(Time.hours(1)))
                .aggregate(new PageViewCountAggFunc(), new PageViewCountWindowFunc());
//        pageViewCountAggDS.print("开窗求和后的结果");

        //根据窗口分组计算每个窗口内的数量
        SingleOutputStreamOperator<PageViewCountBean> totalPageViewCountDS = pageViewCountAggDS
                .keyBy(data -> data.getWindowEnd())
                .process(new PageViewCountProcessFunc());
//        totalPageViewCountDS.print("最终结果");
        //将数据转为指定格式最后sink到数据库
        SingleOutputStreamOperator<PageViewCountResultBean> ResultDS = totalPageViewCountDS.map(line -> {
            LocalDateTime startTime = TimestampTransformUtil.milliTimestampToLocalDateTime(line.getWindowStart());
            LocalDateTime endTime = TimestampTransformUtil.milliTimestampToLocalDateTime(line.getWindowEnd());
            return new PageViewCountResultBean(startTime.toString(), endTime.toString(), line.getCount());
        });
        ResultDS.print("格式化后的结果");
        env.execute("DwdPageView");
    }
}
